Chapter 10 – Introduction to Artificial Neural Networks
This notebook contains all the sample code and solutions to the exercises in chapter 10.
Setup¶
First, let's make sure this notebook works well in both python 2 and 3, import a few common modules, ensure MatplotLib plots figures inline and prepare a function to save the figures:
In [3]:
# To support both python 2 and python 3
from __future__ import division, print_function, unicode_literals
# Common imports
import numpy as np
import os
# to make this notebook's output stable across runs
def reset_graph(seed=42):
tf.reset_default_graph()
tf.set_random_seed(seed)
np.random.seed(seed)
# To plot pretty figures
%matplotlib inline
import matplotlib
import matplotlib.pyplot as plt
plt.rcParams['axes.labelsize'] = 14
plt.rcParams['xtick.labelsize'] = 12
plt.rcParams['ytick.labelsize'] = 12
# Where to save the figures
PROJECT_ROOT_DIR = "."
CHAPTER_ID = "ann"
def save_fig(fig_id, tight_layout=True):
path = os.path.join(PROJECT_ROOT_DIR, "images", CHAPTER_ID, fig_id + ".png")
print("Saving figure", fig_id)
if tight_layout:
plt.tight_layout()
plt.savefig(path, format='png', dpi=300)
命題解決の手法の一つとして「脳の構造を模倣する = 人工ニューラルネット(ANN:artificial neural networks)」の研究が昔からされていた。
10.1.3 Perceptrons¶
In [8]:
from IPython.display import Image
Image("./images/単純パーセプトロン_r2.jpg")
Out[8]:
何ができるの? ... 単純な線形2項分類は一つのLTUでできる。(陰性陽性クラスの分類で、しきい値を超えたら陽。)
ではどのように学習するのか?...生物学的ニューロンは、ニューロンがほかのニューロンを発火するうちに両者の間のつながりが強化されることから、「個々のインスタンスに対して予測を行い、正しい予測を生み出すのに役立ったニューロンからの接続の重みを上げる。」
In [9]:
Image("./images/パーセプトロンの重みの更新.jpg")
Out[9]:
In [10]:
Image("./images/パーセプトロン.jpg")
Out[10]:
Note: we set max_iter and tol explicitly to avoid warnings about the fact that their default value will change in future versions of Scikit-Learn.
上記は、「2個の入力を持ち、インスタンスを同時に3種類の異なるバイナリクラスに分類できる多出力分類器」である。 (入力ニューロンは与えられた入力をそのまま出力)
補足¶
線形分離可能とは、幾何学においてふたつの集合が二次元平面上にあるとき、それらの集合を一本の直線で分離できることをいう。(bi wikipedia)
単一のLTUネットワークを実装する Perceptronを使って、irisデータセットを分類してみる
In [1]:
import numpy as np
from sklearn.datasets import load_iris
from sklearn.linear_model import Perceptron
iris = load_iris()
X = iris.data[:, (2, 3)] # petal length, petal width
y = (iris.target == 0).astype(np.int)
per_clf = Perceptron(max_iter=100, tol=-np.infty, random_state=42)
per_clf.fit(X, y)
y_pred = per_clf.predict([[2, 0.5]])
In [2]:
y_pred
Out[2]:
In [4]:
a = -per_clf.coef_[0][0] / per_clf.coef_[0][1]
b = -per_clf.intercept_ / per_clf.coef_[0][1]
axes = [0, 5, 0, 2]
x0, x1 = np.meshgrid(
np.linspace(axes[0], axes[1], 500).reshape(-1, 1),
np.linspace(axes[2], axes[3], 200).reshape(-1, 1),
)
X_new = np.c_[x0.ravel(), x1.ravel()]
y_predict = per_clf.predict(X_new)
zz = y_predict.reshape(x0.shape)
plt.figure(figsize=(10, 4))
plt.plot(X[y==0, 0], X[y==0, 1], "bs", label="Not Iris-Setosa")
plt.plot(X[y==1, 0], X[y==1, 1], "yo", label="Iris-Setosa")
plt.plot([axes[0], axes[1]], [a * axes[0] + b, a * axes[1] + b], "k-", linewidth=3)
from matplotlib.colors import ListedColormap
custom_cmap = ListedColormap(['#9898ff', '#fafab0'])
plt.contourf(x0, x1, zz, cmap=custom_cmap)
plt.xlabel("Petal length", fontsize=14)
plt.ylabel("Petal width", fontsize=14)
plt.legend(loc="lower right", fontsize=14)
plt.axis(axes)
save_fig("perceptron_iris_plot")
plt.show()
※ロジスティック回帰分類器とは異なり、パーセプトロンは、クラスに属する確率は出力できないことに注意。
パーセプトロンはごく簡単な問題(例えば排他的ORの分類問題)を解けないという課題があった。
その解決方法として、MLP(multi-layer-perception: 多層パーセプトロン)が編み出された。
10.1.4 MLPとバックプロパゲーション¶
MLPは、一つの入力層と隠れ層と呼ばれる一つ以上のLTU層、出力層から構成され、隠れ層が複数ある場合は、深層ニューラルネット(DNN:Depp Neural Network)と呼ばれる。
そして訓練アルゴリズムは、バックプロパゲーション(誤差逆伝搬)と呼ばれる。
バックプロパゲーション...個々の訓練インスタンスに対して、まず予測を行い(前進パス)、誤差を測定してから、各層を後退しながら個々の接続部の誤差への影響おを測定し(後退パス:誤差伝搬)、最後に誤差を最小にするように接続部の重みにわずかな調整を加える(勾配降下ステップ)
このアルゴリズムを正しく動作させるために、ステップ関数をロジスティック関数などの活性化関数に変更した(理由：ステップ関数はフラットな線分で構成されるため、勾配がない=微分しても0になる)
Activation functions¶
In [4]:
def sigmoid(z):
return 1 / (1 + np.exp(-z))
def relu(z):
return np.maximum(0, z)
def derivative(f, z, eps=0.000001):
return (f(z + eps) - f(z - eps))/(2 * eps)
In [5]:
z = np.linspace(-5, 5, 200)
plt.figure(figsize=(11,4))
plt.subplot(121)
plt.plot(z, np.sign(z), "r-", linewidth=1, label="Step")
plt.plot(z, sigmoid(z), "g--", linewidth=2, label="Sigmoid")
plt.plot(z, np.tanh(z), "b-", linewidth=2, label="Tanh")
plt.plot(z, relu(z), "m-.", linewidth=2, label="ReLU")
plt.grid(True)
plt.legend(loc="center right", fontsize=14)
plt.title("Activation functions", fontsize=14)
plt.axis([-5, 5, -1.2, 1.2])
plt.subplot(122)
plt.plot(z, derivative(np.sign, z), "r-", linewidth=1, label="Step")
plt.plot(0, 0, "ro", markersize=5)
plt.plot(0, 0, "rx", markersize=10)
plt.plot(z, derivative(sigmoid, z), "g--", linewidth=2, label="Sigmoid")
plt.plot(z, derivative(np.tanh, z), "b-", linewidth=2, label="Tanh")
plt.plot(z, derivative(relu, z), "m-.", linewidth=2, label="ReLU")
plt.grid(True)
#plt.legend(loc="center right", fontsize=14). 導関数 = 傾き
plt.title("Derivatives", fontsize=14)
plt.axis([-5, 5, -0.2, 1.2])
save_fig("activation_functions_plot")
plt.show()
In [ ]:
def heaviside(z):
return (z >= 0).astype(z.dtype)
def mlp_xor(x1, x2, activation=heaviside):
return activation(-activation(x1 + x2 - 1.5) + activation(x1 + x2 - 0.5) - 0.5)
In [7]:
x1s = np.linspace(-0.2, 1.2, 100)
x2s = np.linspace(-0.2, 1.2, 100)
x1, x2 = np.meshgrid(x1s, x2s)
z1 = mlp_xor(x1, x2, activation=heaviside)
z2 = mlp_xor(x1, x2, activation=sigmoid)
plt.figure(figsize=(10,4))
plt.subplot(121)
plt.contourf(x1, x2, z1)
plt.plot([0, 1], [0, 1], "gs", markersize=20)
plt.plot([0, 1], [1, 0], "y^", markersize=20)
plt.title("Activation function: heaviside", fontsize=14)
plt.grid(True)
plt.subplot(122)
plt.contourf(x1, x2, z2)
plt.plot([0, 1], [0, 1], "gs", markersize=20)
plt.plot([0, 1], [1, 0], "y^", markersize=20)
plt.title("Activation function: sigmoid", fontsize=14)
plt.grid(True)
コラム¶
生物学的ニューロンは、おおよそシグモイドの活性化関数を使っているかのように見えるため、長い間研究されたが、ANNでは一般にReLu活性化関数の方が性能がよいことが分かっており、これは生物学からの類推がうまくいかない例である。
FNN for MNIST¶
MLPは個々の出力がバイナリなので、分類タスクによく使われるが、ロジスティック回帰のようにそれぞれのクラスの確率を出力したい場合は、出力層の活性化関数は、個別の活性化関数でなく、共有のソフトマックス関数を使う。このアーキテクチャは順伝番型ニューラルネット(Feedforward neural network)の例になっている。
Using the Estimator API (formerly tf.contrib.learn)¶
In [8]:
import tensorflow as tf
Warning: tf.examples.tutorials.mnist is deprecated. We will use tf.keras.datasets.mnist instead. Moreover, the tf.contrib.learn API was promoted to tf.estimators and tf.feature_columns, and it has changed considerably. In particular, there is no infer_real_valued_columns_from_input() function or SKCompat class.
In [10]:
(X_train, y_train), (X_test, y_test) = tf.keras.datasets.mnist.load_data()
X_train = X_train.astype(np.float32).reshape(-1, 28*28) / 255.0
X_test = X_test.astype(np.float32).reshape(-1, 28*28) / 255.0
y_train = y_train.astype(np.int32)
y_test = y_test.astype(np.int32)
X_valid, X_train = X_train[:5000], X_train[5000:]
y_valid, y_train = y_train[:5000], y_train[5000:]
In [11]:
feature_cols = [tf.feature_column.numeric_column("X", shape=[28 * 28])]
dnn_clf = tf.estimator.DNNClassifier(hidden_units=[300,100], n_classes=10,
feature_columns=feature_cols)
input_fn = tf.estimator.inputs.numpy_input_fn(
x={"X": X_train}, y=y_train, num_epochs=40, batch_size=50, shuffle=True)
dnn_clf.train(input_fn=input_fn)
Out[11]:
In [12]:
test_input_fn = tf.estimator.inputs.numpy_input_fn(
x={"X": X_test}, y=y_test, shuffle=False)
eval_results = dnn_clf.evaluate(input_fn=test_input_fn)
In [13]:
eval_results
Out[13]:
In [14]:
y_pred_iter = dnn_clf.predict(input_fn=test_input_fn)
y_pred = list(y_pred_iter)
y_pred[0]
Out[14]:
10.3 Using plain TensorFlow¶
ここでは、ネットアーキテクチャを自分で定義した、ミニバッチ勾配降下法でMNISTデータセットを学習したモデルを構築する。
10.3.1 構築フェーズ¶
In [2]:
import tensorflow as tf
n_inputs = 28*28 # MNIST
n_hidden1 = 300 #隠れ層のニューロンの数を指定
n_hidden2 = 100
n_outputs = 10
In [1]:
reset_graph()
# プレースホルダーノードを使って、訓練データと、ターゲットノードを表現する。
# xをプレースホルダーにしたのは、個々の訓練バッチに含まれるインスタンス数がまだ分かっていないから(yも同様で、次元数が1であること以外知らない)
X = tf.placeholder(tf.float32, shape=(None, n_inputs), name="X")
y = tf.placeholder(tf.int32, shape=(None), name="y")
In [17]:
# これから出力層や隠れ層を定義していくにあたり、1つの層を作るための汎用関数を用意する
def neuron_layer(X, n_neurons, name, activation=None):
with tf.name_scope(name):
n_inputs = int(X.get_shape()[1])
stddev = 2 / np.sqrt(n_inputs)
init = tf.truncated_normal((n_inputs, n_neurons), stddev=stddev)
W = tf.Variable(init, name="kernel") #重み行列を格納する
b = tf.Variable(tf.zeros([n_neurons]), name="bias") #バイアス用のための変数.ニューロンごとに一つずつバイアスパラメータが作成
Z = tf.matmul(X, W) + b # Z = X*W + bを計算するサブグラフを作成
if activation is not None:
return activation(Z)
else:
return Z
In [18]:
# 深層ニューラルネットの作成
with tf.name_scope("dnn"):
hidden1 = neuron_layer(X, n_hidden1, name="hidden1",
activation=tf.nn.relu) # 入力層x -> 隠れ層1 .活性化関数はReLu関数
hidden2 = neuron_layer(hidden1, n_hidden2, name="hidden2",
activation=tf.nn.relu) # 隠れ層1 -> 隠れ層2
logits = neuron_layer(hidden2, n_outputs, name="outputs") #ソフトマックス活性化関数に食わせる前の、ネットワークの出力値
In [19]:
# コスト関数(交差エントロピー)を定義する.
# おさらい:交差エントロピーはターゲットクラスに属する確率を低く推定したモデルにペナルティを与える。
with tf.name_scope("loss"):
xentropy = tf.nn.sparse_softmax_cross_entropy_with_logits(labels=y,
logits=logits)
# 出力は各インスタンスの交差エントロピーが収められた一次元テンソル.reduce_mean()により、全てのインスタンスの平均誤差エントロピーを算出.
loss = tf.reduce_mean(xentropy, name="loss")
In [20]:
# 9章で習ったように、モデルパラメータを操作して、コスト関数を最小化するGradientDescentOptimizer(コスト関数から勾配を導いてくれる手法/オプティマイザの一つ)を定義する
learning_rate = 0.01
with tf.name_scope("train"):
optimizer = tf.train.GradientDescentOptimizer(learning_rate)
training_op = optimizer.minimize(loss)
In [21]:
# モデルの評価方法の決定
with tf.name_scope("eval"):
correct = tf.nn.in_top_k(logits, y, 1)
accuracy = tf.reduce_mean(tf.cast(correct, tf.float32))
In [22]:
init = tf.global_variables_initializer()
saver = tf.train.Saver()
10.3.2 実行フェーズ¶
In [23]:
n_epochs = 40
batch_size = 50
In [24]:
def shuffle_batch(X, y, batch_size):
rnd_idx = np.random.permutation(len(X))
n_batches = len(X) // batch_size
for batch_idx in np.array_split(rnd_idx, n_batches):
X_batch, y_batch = X[batch_idx], y[batch_idx]
yield X_batch, y_batch
In [25]:
with tf.Session() as sess:
init.run()
for epoch in range(n_epochs):
for X_batch, y_batch in shuffle_batch(X_train, y_train, batch_size):
sess.run(training_op, feed_dict={X: X_batch, y: y_batch})
acc_batch = accuracy.eval(feed_dict={X: X_batch, y: y_batch})
acc_val = accuracy.eval(feed_dict={X: X_valid, y: y_valid})
print(epoch, "Batch accuracy:", acc_batch, "Val accuracy:", acc_val)
save_path = saver.save(sess, "./my_model_final.ckpt")
In [26]:
with tf.Session() as sess:
saver.restore(sess, "./my_model_final.ckpt") # or better, use save_path
X_new_scaled = X_test[:20]
Z = logits.eval(feed_dict={X: X_new_scaled})
y_pred = np.argmax(Z, axis=1)
In [27]:
print("Predicted classes:", y_pred)
print("Actual classes: ", y_test[:20])
In [28]:
from tensorflow_graph_in_jupyter import show_graph
In [29]:
show_graph(tf.get_default_graph())
Using dense() instead of neuron_layer()¶
Note: previous releases of the book used tensorflow.contrib.layers.fully_connected() rather than tf.layers.dense() (which did not exist when this chapter was written). It is now preferable to use tf.layers.dense(), because anything in the contrib module may change or be deleted without notice. The dense() function is almost identical to the fully_connected() function, except for a few minor differences:
- several parameters are renamed:
scopebecomesname,activation_fnbecomesactivation(and similarly the_fnsuffix is removed from other parameters such asnormalizer_fn),weights_initializerbecomeskernel_initializer, etc. - the default
activationis nowNonerather thantf.nn.relu. - a few more differences are presented in chapter 11.
In [30]:
n_inputs = 28*28 # MNIST
n_hidden1 = 300
n_hidden2 = 100
n_outputs = 10
In [31]:
reset_graph()
X = tf.placeholder(tf.float32, shape=(None, n_inputs), name="X")
y = tf.placeholder(tf.int32, shape=(None), name="y")
In [32]:
with tf.name_scope("dnn"):
hidden1 = tf.layers.dense(X, n_hidden1, name="hidden1",
activation=tf.nn.relu)
hidden2 = tf.layers.dense(hidden1, n_hidden2, name="hidden2",
activation=tf.nn.relu)
logits = tf.layers.dense(hidden2, n_outputs, name="outputs")
y_proba = tf.nn.softmax(logits)
In [33]:
with tf.name_scope("loss"):
xentropy = tf.nn.sparse_softmax_cross_entropy_with_logits(labels=y, logits=logits)
loss = tf.reduce_mean(xentropy, name="loss")
In [34]:
learning_rate = 0.01
with tf.name_scope("train"):
optimizer = tf.train.GradientDescentOptimizer(learning_rate)
training_op = optimizer.minimize(loss)
In [35]:
with tf.name_scope("eval"):
correct = tf.nn.in_top_k(logits, y, 1)
accuracy = tf.reduce_mean(tf.cast(correct, tf.float32))
In [36]:
init = tf.global_variables_initializer()
saver = tf.train.Saver()
In [37]:
n_epochs = 20
n_batches = 50
with tf.Session() as sess:
init.run()
for epoch in range(n_epochs):
for X_batch, y_batch in shuffle_batch(X_train, y_train, batch_size):
sess.run(training_op, feed_dict={X: X_batch, y: y_batch})
acc_batch = accuracy.eval(feed_dict={X: X_batch, y: y_batch})
acc_valid = accuracy.eval(feed_dict={X: X_valid, y: y_valid})
print(epoch, "Batch accuracy:", acc_batch, "Validation accuracy:", acc_valid)
save_path = saver.save(sess, "./my_model_final.ckpt")
In [38]:
show_graph(tf.get_default_graph())
Exercise solutions¶
1. to 8.¶
See appendix A.
9.¶
Train a deep MLP on the MNIST dataset and see if you can get over 98% precision. Just like in the last exercise of chapter 9, try adding all the bells and whistles (i.e., save checkpoints, restore the last checkpoint in case of an interruption, add summaries, plot learning curves using TensorBoard, and so on).
First let's create the deep net. It's exactly the same as earlier, with just one addition: we add a tf.summary.scalar() to track the loss and the accuracy during training, so we can view nice learning curves using TensorBoard.
In [39]:
n_inputs = 28*28 # MNIST
n_hidden1 = 300
n_hidden2 = 100
n_outputs = 10
In [40]:
reset_graph()
X = tf.placeholder(tf.float32, shape=(None, n_inputs), name="X")
y = tf.placeholder(tf.int32, shape=(None), name="y")
In [41]:
with tf.name_scope("dnn"):
hidden1 = tf.layers.dense(X, n_hidden1, name="hidden1",
activation=tf.nn.relu)
hidden2 = tf.layers.dense(hidden1, n_hidden2, name="hidden2",
activation=tf.nn.relu)
logits = tf.layers.dense(hidden2, n_outputs, name="outputs")
In [42]:
with tf.name_scope("loss"):
xentropy = tf.nn.sparse_softmax_cross_entropy_with_logits(labels=y, logits=logits)
loss = tf.reduce_mean(xentropy, name="loss")
loss_summary = tf.summary.scalar('log_loss', loss)
In [43]:
learning_rate = 0.01
with tf.name_scope("train"):
optimizer = tf.train.GradientDescentOptimizer(learning_rate)
training_op = optimizer.minimize(loss)
In [44]:
with tf.name_scope("eval"):
correct = tf.nn.in_top_k(logits, y, 1)
accuracy = tf.reduce_mean(tf.cast(correct, tf.float32))
accuracy_summary = tf.summary.scalar('accuracy', accuracy)
In [45]:
init = tf.global_variables_initializer()
saver = tf.train.Saver()
Now we need to define the directory to write the TensorBoard logs to:
In [46]:
from datetime import datetime
def log_dir(prefix=""):
now = datetime.utcnow().strftime("%Y%m%d%H%M%S")
root_logdir = "tf_logs"
if prefix:
prefix += "-"
name = prefix + "run-" + now
return "{}/{}/".format(root_logdir, name)
In [47]:
logdir = log_dir("mnist_dnn")
Now we can create the FileWriter that we will use to write the TensorBoard logs:
In [48]:
file_writer = tf.summary.FileWriter(logdir, tf.get_default_graph())
Hey! Why don't we implement early stopping? For this, we are going to need to use the validation set.
In [49]:
m, n = X_train.shape
In [50]:
n_epochs = 10001
batch_size = 50
n_batches = int(np.ceil(m / batch_size))
checkpoint_path = "/tmp/my_deep_mnist_model.ckpt"
checkpoint_epoch_path = checkpoint_path + ".epoch"
final_model_path = "./my_deep_mnist_model"
best_loss = np.infty
epochs_without_progress = 0
max_epochs_without_progress = 50
with tf.Session() as sess:
if os.path.isfile(checkpoint_epoch_path):
# if the checkpoint file exists, restore the model and load the epoch number
with open(checkpoint_epoch_path, "rb") as f:
start_epoch = int(f.read())
print("Training was interrupted. Continuing at epoch", start_epoch)
saver.restore(sess, checkpoint_path)
else:
start_epoch = 0
sess.run(init)
for epoch in range(start_epoch, n_epochs):
for X_batch, y_batch in shuffle_batch(X_train, y_train, batch_size):
sess.run(training_op, feed_dict={X: X_batch, y: y_batch})
accuracy_val, loss_val, accuracy_summary_str, loss_summary_str = sess.run([accuracy, loss, accuracy_summary, loss_summary], feed_dict={X: X_valid, y: y_valid})
file_writer.add_summary(accuracy_summary_str, epoch)
file_writer.add_summary(loss_summary_str, epoch)
if epoch % 5 == 0:
print("Epoch:", epoch,
"\tValidation accuracy: {:.3f}%".format(accuracy_val * 100),
"\tLoss: {:.5f}".format(loss_val))
saver.save(sess, checkpoint_path)
with open(checkpoint_epoch_path, "wb") as f:
f.write(b"%d" % (epoch + 1))
if loss_val < best_loss:
saver.save(sess, final_model_path)
best_loss = loss_val
else:
epochs_without_progress += 5
if epochs_without_progress > max_epochs_without_progress:
print("Early stopping")
break
In [51]:
os.remove(checkpoint_epoch_path)
In [52]:
with tf.Session() as sess:
saver.restore(sess, final_model_path)
accuracy_val = accuracy.eval(feed_dict={X: X_test, y: y_test})
In [53]:
accuracy_val
Out[53]: